package main

import (
	"fmt"
	"math/rand"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/Shopify/sarama"
	//"github.com/bsm/sarama-cluster" //support automatic consumer-group rebalancing and offset tracking
	"github.com/sdbaiguanghe/glog"
)

var (
	topics = "mytopic"
)

// syncProducer 同步生产者
// 并发量小时，可以用这种方式
func syncProducer(str string) {
	config := sarama.NewConfig()
	//  config.Producer.RequiredAcks = sarama.WaitForAll
	//  config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.Producer.Return.Successes = true
	config.Producer.Timeout = 5 * time.Second
	p, err := sarama.NewSyncProducer(strings.Split("localhost:9092", ","), config)
	defer p.Close()
	if err != nil {
		glog.Errorln(err)
		return
	}

	v := "sync消息队列: " + strconv.Itoa(rand.New(rand.NewSource(time.Now().UnixNano())).Intn(10000))
	if str != "" {
		v = str
	}
	fmt.Fprintln(os.Stdout, v)
	msg := &sarama.ProducerMessage{
		Topic: topics,
		Value: sarama.ByteEncoder(v),
	}
	if _, _, err := p.SendMessage(msg); err != nil {
		glog.Errorln(err)
		return
	}
}

// asyncProducer 异步生产者
// 并发量大时，必须采用这种方式
func asyncProducer(str string) {
	config := sarama.NewConfig()
	config.Producer.Return.Successes = true //必须有这个选项
	config.Producer.Timeout = 5 * time.Second
	p, err := sarama.NewAsyncProducer(strings.Split("localhost:9092", ","), config)
	defer p.Close()
	if err != nil {
		return
	}

	//必须有这个匿名函数内容
	go func(p sarama.AsyncProducer) {
		errors := p.Errors()
		success := p.Successes()
		for {
			select {
			case err := <-errors:
				if err != nil {
					glog.Errorln(err)
				}
			case <-success:
			}
		}
	}(p)

	v := "async: " + strconv.Itoa(rand.New(rand.NewSource(time.Now().UnixNano())).Intn(10000))
	if str != "" {
		v = str
	}
	fmt.Fprintln(os.Stdout, v)
	msg := &sarama.ProducerMessage{
		Topic: topics,
		Value: sarama.ByteEncoder(v),
	}
	p.Input() <- msg
}

func main() {
	fmt.Println("please input your message:")
	var msg, msg2 string
	for {
		fmt.Scanf("%s %s", &msg, &msg2)
		syncProducer(time.Now().Format("2006-01-02 15:04.000") + "同步发送:" + msg)
		asyncProducer(time.Now().Format("2006-01-02 15:04.000") + "异步发送:" + msg)
	}
}
